package com.tc.vms.kafka;

import com.tc.vms.log.VmsClientLogger;
import org.slf4j.Logger;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by yonghua.zhang on 2016/5/16.
 */
public abstract class KafkaShutdownableThread extends Thread {
    protected final static Logger LOGGER = VmsClientLogger.getLog();
    protected final AtomicBoolean isRunning = new AtomicBoolean(true);
    protected final CountDownLatch shutdownLatch = new CountDownLatch(1);
    protected final String name;
    protected final boolean isInterruptible;

    public KafkaShutdownableThread(String name, boolean isInterruptible) {
        super(name);
        this.setDaemon(false);
        this.name = name;
        this.isInterruptible = isInterruptible;
    }

    public boolean initiateShutdown() {
        if (isRunning.compareAndSet(true, false)) {
            LOGGER.info("[{}] thread Shutdown Shutting down", name);
            isRunning.set(false);
            if (isInterruptible)
                interrupt();
            return true;
        } else
            return false;
    }

    public void shutdown() {
        initiateShutdown();
        awaitShutdown();
    }

    public void awaitShutdown() {
        try {
            shutdownLatch.await();
            LOGGER.info("[{}] thread Shutdown completed", name);
        } catch (InterruptedException e) {
            LOGGER.error("[{}] shutdown interupted", name);

        }
    }

    abstract public void doWork();

    @Override
    public void run() {
        LOGGER.info("[{}] thread Starting ", name);
        try {
            while (isRunning.get()) {
                doWork();
            }
        } catch (Exception e) {
            if (isRunning.get())
                LOGGER.error("Error due to ", e);
        }
        shutdownLatch.countDown();
        LOGGER.info("[{}] thread Stopped ", name);
    }
}
